package cn.whuc.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.io

object Demo_CategoryTop10_3 {
  def main(args: Array[String]): Unit = {
    // 1 创建sparkContext
    val sc: SparkContext = new SparkContext(
      new SparkConf()
        .setMaster("local[*]")
        .setAppName(" ")
    )

    // 2 编写代码
    // 2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
    //2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_48_2019-07-17 00:00:10_null_16_98_null_null_null_null_19

    val datas: RDD[String] = sc.textFile("input/user_visit_action.txt")
    /*
      一行一种行为
      每个品类 点击 下单 支付
      思路3 处理每一行时 直接处理
        品类 点击次数，0，0
        品类 0，下单次数，0
        品类 0，0，支付次数
        思路1 2  reduceByKey次数过多
        提高计算效率
      排序
      前10
      ok
      考虑细节： 点击一次处理 下单和支付都是2次处理 在处理行数据的时，使用flatMap
      对于点击事件 需要将数据封装List 而下单和支付正常处理
     */
    val result: Array[(String, (Int, Int, Int))] = datas.flatMap(line => {
      val strings: Array[String] = line.split("_")
      if (strings(6) != "-1") {
        List((strings(6), (1, 0, 0)))
      } else if (strings(8) != "null") {
        val ids: Array[String] = strings(8).split(",")
        ids.map(id => {
          (id, (0, 1, 0))
        })
      } else if (strings(10) != "null") {
        val ids: Array[String] = strings(10).split(",")
        ids.map(id => {
          (id, (0, 0, 1))
        })
      } else {
        // 相当于之前思路中的过滤的步骤
        Nil
      }
    }).reduceByKey((t1, t2) => {
      (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
    }).sortBy(_._2, false).take(10)

    result.foreach(println)

    // 3 关闭上下文对象
    sc.stop()
  }

}
